package com.stormdemo.wc;

import java.io.File;

import com.stormdemo.wc.bolt.TransferBolt;
import com.stormdemo.wc.bolt.WorldCountBolt;
import com.stormdemo.wc.bolt.WriteBolt;
import com.stormdemo.wc.spout.WordReadSpout;

import backtype.storm.Config;
import backtype.storm.StormSubmitter;
import backtype.storm.generated.AlreadyAliveException;
import backtype.storm.generated.InvalidTopologyException;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.tuple.Fields;

public class RunMain {

	public static void main(String[] args) {
		TopologyBuilder builder = new TopologyBuilder();
		
		builder.setSpout("input", new WordReadSpout(),4);
		builder.setBolt("transfss", new  TransferBolt(),4).shuffleGrouping("input");
		builder.setBolt("wc", new WorldCountBolt(),4).fieldsGrouping("transfss", new Fields("transf"));
		builder.setBolt("write", new WriteBolt(),4).fieldsGrouping("wc", new Fields("word"));
		
		Config conf = new Config();
		conf.setNumWorkers(2);
		try {
			StormSubmitter.submitTopology("wcCount14", conf, builder.createTopology());
		} catch (AlreadyAliveException e) {
			e.printStackTrace();
		} catch (InvalidTopologyException e) {
			e.printStackTrace();
		}
	}

}
